package com.lanyou.cook.service;

import java.util.concurrent.TimeUnit;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;

import org.apache.commons.lang3.BooleanUtils;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.retry.RetryNTimes;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import com.lanyou.cook.common.Constants;

/*
 * 与ZooKeeper通讯的基础设施
 */

@Service
public class CuratorService {
	private CuratorFramework client;
	private ConnectionState lastConnectionState = ConnectionState.LOST;
	
	@Value("${zookeeper.enable}")
	private boolean isEnable;
	@Value("${zookeeper.connectString}")
	private String zkConnectString;
	
	@PostConstruct
	private void init(){
		if(BooleanUtils.isNotTrue(isEnable))
			return;
		client = CuratorFrameworkFactory.builder()  
		        .connectString(zkConnectString).namespace(Constants.ZOOKEPPER_NAMESPACE)  
		        .retryPolicy(new RetryNTimes(Integer.MAX_VALUE, 1000))  
		        .connectionTimeoutMs(10000).build();
		
		client.getConnectionStateListenable().addListener((client2, newState) -> {
			lastConnectionState = newState;
		});
		
		client.start();
		
		// 启动时连接zookeeper失败，则整个程序启动失败
		try {
			if(!client.blockUntilConnected(30, TimeUnit.SECONDS))
				throw new Error("无法连接到ZooKeeper!");
		} catch (InterruptedException e1) {
			Thread.currentThread().interrupt();
		}
	}
	
	@PreDestroy
	private void close(){
		closeClient();
	}
	
	private void closeClient(){
		if(client != null){
			client.close();
			client = null;
		}
	}
	
	public boolean isEnable() {
		return BooleanUtils.isTrue(isEnable);
	}

	public CuratorFramework getClient(){
		return client;
	}
	
	public boolean isConnected(){
		return lastConnectionState.isConnected();
	}
}
